package com.thiswhyme.mq.kafka.consumer1.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;

/**
 * @description: Kafka配置
 * @author: thiswhyme
 * @date: 2021/4/21
 */
@Slf4j
@Configuration
public class KafkaConfig {

    @Value("${spring.application.name}")
    private String appName;

    // 新建一个异常处理器，用@Bean注入
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return (message, exception, consumer) -> {
            System.out.println("消费异常："+message.getPayload());
            return null;
        };
    }

    /**
     * 自定义KafkaListenerContainerFactory，用于过滤消息
     * @param consumerFactory
     * @return
     */
    @Bean("filterContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> filterContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);

        // 消息过滤策略，返回true消息则被过滤
        factory.setRecordFilterStrategy(consumerRecord -> {
            log.info("====进入消息过滤器, {}", consumerRecord.toString());
            return !appName.equals(consumerRecord.key());
        });
        // 被过滤的消息是否丢弃
        factory.setAckDiscarded(true);
        return factory;
    }

    /**
     * 自定义KafkaListenerContainerFactory，用于禁止KafkaListener自启
     * @param consumerFactory
     * @return
     */
    @Bean("delayContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> delayContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);

        //禁止KafkaListener自启动
        factory.setAutoStartup(false);
        return factory;
    }
}
